package com.algo.utils

import org.apache.spark._
import org.apache.spark.SparkContext._
import org.apache.spark.rdd.RDD

/**
 * this class deals with panel
 * @author lujianfeng@miaozhen.com
 *
 */
object PanelUtils{

  val sexClasses = 2
  val ageClasses = 8
  val eduClasses = 4
  val salaryClasses = 5

  def getPanel(sc: SparkContext, path: String): RDD[(String, String)] = {
    sc.textFile(path)
      .flatMap(line => {
        val pos = line.indexOf(",")
        if(pos > 0)
                Iterator((line.substring(0, pos), line.substring(pos+1)))
        else    Iterator()
       })
  }
  def parsePanelLabel(items: Array[String]): List[Int] = {
    var sex = 0;
    if(items(2).equals("3")) sex = 1
    var age = 0
    if(items(3).equals("6")) age = 1
    else if(items(3).equals("7")) age = 2
    else if(items(3).equals("8")) age = 3
    else if(items(3).equals("9")) age = 4
    else if(items(3).equals("10")) age = 5
    else if(items(3).equals("11")) age = 6
    else if(items(3).equals("41")) age = 7

    var edu = 0
    if(items(4).equals("17")) edu = 1
    else if(items(4).equals("18")) edu = 2
    else if(items(4).equals("42")) edu = 3

    var salary = 0
    if(items(5).equals("22")) salary = 1
    else if(items(5).equals("23")) salary = 2
    else if(items(5).equals("24")) salary = 3
    else if(items(5).equals("25")) salary = 4

    List(sex, age, edu, salary)
  }
  def parseExtendedPanelLabel(items: Array[String]): List[Int] = {
      var sex = 0
      if(items(3).equals("1")) sex = 1
      var age = 0;
      for(i <- 5 to 11)
        if(items(i).equals("1"))
          age = i - 4
      var edu = 0;
      for(i <- 13 to 15)
        if(items(i).equals("1"))
          edu = i - 12
      var salary = 0
      for(i <- 17 to 20)
        if(items(i).equals("1"))
          salary = i - 16

      List(sex, age, edu, salary)
  }
  def addPopAttr(count: Array[Int], items: Array[String]){
    val list = parsePanelLabel(items)
    count(list(0)) += 1
    count(sexClasses + list(1)) += 1
    count(sexClasses + ageClasses + list(2)) += 1
    count(sexClasses + ageClasses + eduClasses + list(3)) += 1
  }

}
